package com.huawei.app.demo.video;

import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ArrayBlockingQueue;

import com.huawei.app.demo.util.LogUtils;
import com.huawei.app.httpapi.ThreadPools;

/**
 * 数据接收线程
 *
 * @since 2020-08-14
 */
public class ReceiverTask implements Runnable {

    private final static String TAG = ReceiverTask.class.getSimpleName();

    private String host;
    private int port;
    private boolean isRuning = true;
    private boolean connectOk = false;

    private ServerSocket server;
    private Socket socket;
    private InputStream inputStream;
    private ObjectInputStream objInputStream;
    private ArrayBlockingQueue<ImgBuffer> receiverQueue;

    public ReceiverTask(String host, int port, ArrayBlockingQueue<ImgBuffer> receiverQueue) {
        this.host = host;
        this.port = port;
        this.receiverQueue = receiverQueue;

        init();
    }

    private void init() {
        try {
            InetAddress localAddress = InetAddress.getByName(host);
            this.server = new ServerSocket(port, 50, localAddress);
            LogUtils.getInstance().i(TAG, "init socket server on " + host + ":" + port + " success.");
        } catch (IOException e) {
            LogUtils.getInstance().e(TAG, "init socket server on " + host + ":" + port + " failed, " + e.getMessage());
        }
    }

    private void initSocket() {
        try {
            // 关闭上次资源
            close();

            // 重新初始化连接
            if (this.server == null) {
                init();
            }
            this.socket = server.accept();
            this.socket.setKeepAlive(true);
            this.socket.setReuseAddress(true);
            this.socket.setSoLinger(true, 10000);
            this.socket.setSoTimeout(120000);

            this.inputStream = socket.getInputStream();
            this.objInputStream = new ObjectInputStream(inputStream);
            this.connectOk = true;
            LogUtils.getInstance().i(TAG, "init socket server accept " + socket.getRemoteSocketAddress() + " success.");
        } catch (IOException e) {
            LogUtils.getInstance().e(TAG, "init socket server accept " + socket.getRemoteSocketAddress() + " failed, " + e.getMessage());
        }
    }

    @Override
    public void run() {
        String oldThreadName = Thread.currentThread().getName();
        Thread.currentThread().setName(this.getClass().getSimpleName());

        ImgBuffer imgBuffer = null;
        while (isRuning) {
            try {
                if (!isConnected()) {
                    initSocket();
                    if (!isConnected()) {
                        ThreadPools.pause(1000);
                        continue;
                    }
                }
                ThreadPools.pause(20);

                if (imgBuffer == null) {
                    LogUtils.getInstance().v(TAG, "before objInputStream.readObject");
                    imgBuffer = (ImgBuffer)this.objInputStream.readObject();
                    if (imgBuffer == null) {
                        ThreadPools.pause(20);
                        continue;
                    }
                }

                LogUtils.getInstance().v(TAG, "before receiverQueue.offer sno: " + imgBuffer.getSno() + ", snoIndex: " + imgBuffer.getSnoIndex());
                this.receiverQueue.offer(imgBuffer);
                LogUtils.getInstance().v(TAG, "after receiverQueue.offer sno: " + imgBuffer.getSno() + ", snoIndex: " + imgBuffer.getSnoIndex());
                imgBuffer = null;
            } catch (Exception e) {
                this.connectOk = false;
                LogUtils.getInstance().e(TAG, e.getMessage());
            }
        }

        Thread.currentThread().setName(oldThreadName);
    }

    private boolean isConnected() {
        return (this.socket != null && this.socket.isConnected() && !this.socket.isClosed() && this.connectOk);
    }

    public void stop() {
        this.isRuning = false;
        close();
        if (this.server != null) {
            closeQuietly(this.server);
        }
    }

    private void close() {
        closeQuietly(this.objInputStream);
        closeQuietly(this.inputStream);
        closeQuietly(this.socket);
    }

    private void closeQuietly(Closeable io) {
        try {
            if (io != null) {
                io.close();
            }
        } catch (IOException e) {
            // ignore.
        }
    }
}

